package com.xiaopengpeng.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import com.xiaopengpeng.api.HbaseOperations;
import com.xiaopengpeng.api.RowMapper;
import com.xiaopengpeng.api.TableCallback;
@Service
public class HbaseTemplate implements HbaseOperations{
	Logger logger = LoggerFactory.getLogger(HbaseTemplate.class);
	
	private Connection connection = HbaseConnetcion.getConnection();
	
	
	
	@Override
	public <T> T execute(String tableName, TableCallback<T> action) {
		Table table = null;
		try {
			connection.getTable(TableName.valueOf(tableName));
			return action.doInTable(table);
		} catch (Exception e) {
			logger.error(e.toString());
		}finally {
			if(table!=null){
				try {
					table.close();
				} catch (IOException e) {
					logger.error(e.toString());
				}
			}
		}
		return null;
	}

	@Override
	public <T> List<T> find(String tableName, String family, RowMapper<T> mapper) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public <T> List<T> find(String tableName, String family, String qualifier, RowMapper<T> mapper) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public <T> List<T> find(String tableName, Scan scan, RowMapper<T> action) {
		
        return this.execute(tableName, new TableCallback<List<T>>() {
            @Override
            public List<T> doInTable(Table table) throws Exception{
                int caching = scan.getCaching();
                // 如果caching未设置(默认是1)，将默认配置成5000
                if (caching == 1) {
                    scan.setCaching(5000);
                }
                ResultScanner scanner = table.getScanner(scan);
                try {
                    List<T> rs = new ArrayList<T>();
                    int rowNum = 0;
                    for (Result result : scanner) {
                        rs.add(action.mapRow(result, rowNum++));
                    }
                    return rs;
                } finally {
                    scanner.close();
                }
            }
        });
	}

}
